import datetime
import json
import pytz
import time
import signal

import google.protobuf.json_format as json_format

from tornado.ioloop import IOLoop
from absl import app, flags

from concurrent.futures import ProcessPoolExecutor, as_completed
from coin.util.queue.config import KafkaConfig
from coin.support.proto_log.app.strat_state_anaylyzer_v2 import AppConfig, DataEngine
from cc.experimental.wliang.ksqldb.ksqldb_client import KafkaClient


FLAGS = flags.FLAGS


def _get_today():
  return datetime.date.today().strftime('%Y%m%d')

def _get_yesterday():
  return (datetime.date.today() - datetime.timedelta(days=1)).strftime('%Y%m%d')


def _get_min_max_dt(trading_date):
  return (
      datetime.datetime.combine(trading_date, datetime.time.min.replace(tzinfo=pytz.UTC)),
      datetime.datetime.combine(trading_date, datetime.time.max.replace(tzinfo=pytz.UTC))
  )


class Destributor(object):
  def __init__(self, ksql_client):
    self.ksql_client = ksql_client

  def _filter(self, log):
    return False

  def process(self, topic, timestamp, log):
    if self._filter(log):
      return;
    data = json_format.MessageToDict(log, preserving_proto_field_name=True)
    rows = []
    rows.append(json.dumps(data).encode('utf-8'))
    self.ksql_client.produce(rows, topic)


def _produce(kafka_server, kafka_topics, bootstrap_server):
  app_config = AppConfig(og_log_config=None,
                         kafka_config=KafkaConfig(kafka_servers=[kafka_server]),
                         plotter_config=None)
  ksql_client = KafkaClient(bootstrap_server)
  destributor = Destributor(ksql_client)
  engine = DataEngine(app_config, destributor, None, None, kafka_topics)

  ioloop = IOLoop.current()
  engine.start(ioloop=ioloop)
  signal.signal(signal.SIGTERM, engine.signal)
  signal.signal(signal.SIGINT, engine.signal)

  try:
    ioloop.start()
  except (KeyboardInterrupt, SystemExit):
    print("%s exception" % ','.join(kafka_topics))
    return

def main(_):
  assert(FLAGS.kafka_topics)
  _produce(kafka_server = FLAGS.kafka_server,
           kafka_topics = FLAGS.kafka_topics.split(','),
           bootstrap_server = FLAGS.bootstrap_server)


if __name__ == '__main__':
  flags.DEFINE_string('bootstrap_server', "localhost:9092", "bootstrap_server")
  flags.DEFINE_string('kafka_server', "coin-kafka.corp.prestolabs.io:9092", "kafka server")
  flags.DEFINE_string('kafka_topics', None, "kafka topics")
  app.run(main)
